package com.bw.gmall.realtime.dws.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import com.bw.gmall.realtime.dws.function.SplitFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Executable;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow extends BaseSQLApp {
    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(10021,4, "dws_traffic_source_keyword_page_view_window");
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {
        TableConfig config=tEnv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","5s");

        // 1. 读取 页面日志
        create_page(tEnv);
        tEnv.sqlQuery("select * from page_log");


        // 2. 读取搜索关键词
        Excepage(tEnv);

        //注册 函数
        tEnv.createTemporaryFunction("SplitFunction", SplitFunction.class);

        // 3. 自定义分词函数
        fencistream1(tEnv);

        // 4. 开窗聚和 tvf

        Table table = JujiStream(tEnv);

        // 5. 写出到 doris 中
        create_dorie(tEnv);

        table.insertInto(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW).execute();


    }

    private static void create_dorie(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table dws_traffic_source_keyword_page_view_window \n" +
                "(\n" +
                " stt string,\n" +
                " edt string,\n" +
                " cur_date string,\n" +
                " keyword string,\n" +
                " keyword_count bigint\n" +
                ")" + SQLUtil.getDorisSinkSQL(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW));
    }

    private static Table JujiStream(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("SELECT\n" +
                "  date_format(TUMBLE_START(et, interval '5' second),'yyyy-MM-dd HH:mm:ss') AS stt,\n" +
                "  date_format(TUMBLE_END(et, interval '5' second),'yyyy-MM-dd HH:mm:ss') AS edt,\n" +
                "  date_format(now() , 'yyyyMMdd') cur_date, " +
                "  keyword,\n" +
                "  count(*) keyword_count\n" +
                "FROM page_info1\n" +
                "GROUP BY\n" +
                "  TUMBLE(et, interval '5' second ),\n" +
                "  keyword");

//        Table table = tEnv.sqlQuery("select " +
//                " date_format(window_start,'yyyy-MM-dd HH:mm:ss') as `stt`," +
//                " date_format(window_end,'yyyy-MM-dd HH:mm:ss') as `edt`," +
//                " cast(date_format(now(), 'yyyyMMdd') as string) cur_date," +
//                " keyword," +
//                " count(*) keyword_count," +
//                " UNIX_TIMESTAMP()*1000 ts"+
//                " from " +
//                " TABLE( TUMBLE(TABLE page_info1, DESCRIPTOR(et), INTERVAL '5' second)) " +
//                " GROUP BY window_start, window_end,keyword ");

//        tEnv.createTemporaryView("page_stream",table2);
    }

    private static void fencistream1(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "item,\n" +
                "et,\n" +
                "word keyword \n" +
                "from page_info,lateral table(SplitFunction(item))");

        tEnv.createTemporaryView("page_info1",table);
    }

    private static void Excepage(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "`page`['item'] item,\n" +
                " et\n" +
                "from page_log\n" +
                "where `page`['item_type']='keyword'\n" +
                "and `page`['last_page_id']='search'");

        tEnv.createTemporaryView("page_info",table);
    }

    private static void create_page(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table page_log\n" +
                "(\n" +
                " page map<string,string>,\n" +
                " ts bigint,\n" +
                " et as TO_TIMESTAMP_LTZ(ts,3),\n" +
                " WATERMARK FOR et AS et - INTERVAL '5' SECOND\n" +
                ")" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.TOPIC_DWD_TRAFFIC_PAGE));
    }
}
